/**
 * Software License, Version 1.0
 *
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 *
 *
 *Redistribution and use in source and binary forms, with or without
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 *
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University.  For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering,
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 *
 *
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of
 * any other entity.  Indiana University disclaims any liability to any
 * recipient for claims brought by any other entity based on infringement of
 * intellectual property rights or otherwise.
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.reliable.impl;

import java.util.Hashtable;
import java.util.LinkedList;

import cgl.narada.event.EventHeaders;
import cgl.narada.event.EventID;
import cgl.narada.event.NBEvent;
import cgl.narada.event.NBEventTypes;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.event.impl.NBEventGenerator;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.reliable.events.RdCompanionEntityEvent;
import cgl.narada.service.reliable.events.RdNegotiationsServiceEvent;
import cgl.narada.service.reliable.events.RdPublisherRecoveryInfo;
import cgl.narada.service.reliable.events.RdRecoveryEntityRequest;
import cgl.narada.service.reliable.events.RdRecoveryResponse;
import cgl.narada.service.reliable.events.RdRepublishedEntityEvent;

/** Keeps track of published events that have been archived onto stable
    storage and those that haven't been. Also facilitates retransmission of
    events upon receipt of a

    @author Shrideep Pallickara and Hasan Bulut
    $Date$
    $Revision$
 */


public class PublisherPercolator implements RdDebugFlags {

    private String moduleName = "PublisherPercolator: ";

    /** Keeps a list of events that have still not been archived.
        EventID --> NBEvent */
    private Hashtable pendingEvents;

    /** Keeps the list of the companion events associated with the campanions
        EventID --> CompanionEvent */
    private Hashtable pendingCompanions;


    /** */
    private Hashtable templateCatenationChain;

    /** Keeps a list of the last catenations associated with the templateId
        in question. TemplateId --> catenation */
    private Hashtable templatesAndCatenations;


    /** Keeps a list of the last acknowledged catenations associated with
        various templates. TemplateId --> catenation */
    private Hashtable acknowledgedCatenations;

    /** Keeps track of the catenation numbers that need to assiagned to published
        events */
//    private int assignedCatenation = 0;
    private Hashtable assignedCatenations = new Hashtable();


    private long lastKnownTimestamp = 0;
    private int lastKnownSequencing = 0;
    private int entityId;

    /** Keeps track of both the catenation number, and the event id associated
        with the catenation number. */

    /**CatenationNumber --> EventID */
    private Hashtable catenationAndEventIds;

    private EventProducer producer;

    private String rdsString = "ReliableDeliveryService/";

    private RdFragmentationHelper fragmentationHelper;

    public PublisherPercolator(int entityId, EventProducer producer) {
        this.entityId = entityId;
        this.producer = producer;

        pendingEvents = new Hashtable();
        pendingCompanions = new Hashtable();
        templateCatenationChain = new Hashtable();
        templatesAndCatenations = new Hashtable();
        acknowledgedCatenations = new Hashtable();

        catenationAndEventIds = new Hashtable();
    }


    public void
            setFragmentationHelper(RdFragmentationHelper fragmentationHelper) {
        this.fragmentationHelper = fragmentationHelper;
    }


    public void publishEvent(NBEvent nbEvent) throws ServiceException {
        EventID eventId = nbEvent.getEventHeaders().getEventId();
        int templateId = nbEvent.getEventHeaders().getTemplateId();
        if (shouldBeDiscarded(eventId)) {
            return;
        }

        RdCompanionEntityEvent companionEvent = createCompanionEvent(nbEvent);
        Object catenationKey = "" + templateId +
                               companionEvent.getCatenationNumber();

        pendingEvents.put(eventId, nbEvent);
        pendingCompanions.put(eventId, companionEvent);
        catenationAndEventIds.put(catenationKey, eventId);

        if (PublisherPercolator_Debug) {
            System.out.println(moduleName + "Publishing event with id" +
                               eventId);
        }
        producer.publishEvent(nbEvent);

        NBEvent nbEvent2 =
                producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
                                       rdsString + templateId,
                                       companionEvent.getBytes());

        NBEventGenerator.setEventType(nbEvent2,
                                      NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);

        producer.publishEvent(nbEvent2);

        if (PublisherPercolator_Debug) {
            System.out.println(moduleName + "Publishing companion event for id" +
                               eventId);
            System.out.println(moduleName +
                               "First phase of reliable publish over " +
                               "CATENATION NUMBER = [" + catenationKey + "]");
        }

        /** We need to make sure that nbEvent has already been serialized and
          scheduled for delivery, before we proceed to drain the payload in
          case the event in question is a fragmented event.
         When the communications service runs in its own thread we need to get
         the bytes for events to be sent across, place it in the queue and then
          return. This would ensure that subsequent changes to nbEvent will not
          mean that the changes are what would be sent across.*/
        if (isFragmented(nbEvent)) {
            /* The draining needs to be done to ensure that while we are sending
               extremely large payloads, the memory utilization does not cause
               an OutOfMemory Error */
            try {
                int _catNum = companionEvent.getCatenationNumber();
                fragmentationHelper.drainPayloadFromEvent(nbEvent);
                fragmentationHelper.catenationNumberAssigned(nbEvent, _catNum);
            } catch (ServiceException serEx) {
                System.out.println(moduleName + serEx);
            }
        }

    }


    /** Based on the event id's contained in the event, determine if we should
        discard it since it should have been processed before */
    private boolean shouldBeDiscarded(EventID eventId) {
        long _timestamp = eventId.getTimeStamp();
        int _seqNum = eventId.getSequenceNumber();

        if (_timestamp < lastKnownTimestamp) {
            return true;
        }

        if (_timestamp > lastKnownTimestamp) {
            lastKnownTimestamp = _timestamp;
            lastKnownSequencing = _seqNum;
            return false;
        }

        /** The time stamps obviously should match now. */

        /** If seqnum is greater its not a duplicate otherwise it is */
        if (_seqNum > lastKnownSequencing) {
            lastKnownSequencing = _seqNum;
            return false;
        }
        return true;
    }

    /** Create a companion event for the nbEvent */
    private RdCompanionEntityEvent createCompanionEvent(NBEvent nbEvent) {
        EventID eventId = nbEvent.getEventHeaders().getEventId();
        int templateId = nbEvent.getEventHeaders().getTemplateId();

        Object templateKey = new Integer(templateId);

        /** If an entry does not exist create one */
        if (!templatesAndCatenations.containsKey(templateKey)) {
            templatesAndCatenations.put(templateKey, new Integer(0));
            templateCatenationChain.put(templateKey, new LinkedList());
        }

        int _previousCatenation =
                ((Integer) templatesAndCatenations.get(templateKey)).intValue();

        int assignedCatenation = ((Integer) assignedCatenations.get(templateKey)).
                                 intValue();
        assignedCatenation++;

        assignedCatenations.remove(templateKey);
        assignedCatenations.put(templateKey, new Integer(assignedCatenation));

        RdCompanionEntityEvent companionEvent =
                new RdCompanionEntityEvent(entityId, templateId,
                                           assignedCatenation,
                                           _previousCatenation, eventId);

        templatesAndCatenations.remove(templateKey);
        templatesAndCatenations.put(templateKey, new Integer(assignedCatenation));

        LinkedList _catenationChain = (LinkedList) templateCatenationChain.get(
                templateKey);
        _catenationChain.add(new Integer(assignedCatenation));
        return companionEvent;
    }


    public void
            processNegotiationsEvent(RdNegotiationsServiceEvent
                                     negotiationEvent) {
        if (!negotiationEvent.isAck()) {
            republishEvent(negotiationEvent);
        } else {
            processAcknowledgement(negotiationEvent);
        }
    }


    private void republishEvent(RdNegotiationsServiceEvent negotiationEvent) {
        int catenationNum = negotiationEvent.getNakCatenationNumber();
        int templateId = negotiationEvent.getTemplateId();
        Object catenationKey = "" + templateId + catenationNum;

        if (!catenationAndEventIds.containsKey(catenationKey)) {
            System.out.println(moduleName +
                               "Incorrect request for retransmission " +
                               "of catenation number = " + catenationNum);
            return;
        }

        if (PublisherPercolator_Debug) {
            System.out.println(moduleName +
                               "Retransmission catenation number = " +
                               catenationNum + "\n\n");
        }

        Object eventId = catenationAndEventIds.get(catenationKey);

        NBEvent nbEvent = (NBEvent) pendingEvents.get(eventId);

        /** If the event is a fragmented one, we need to refill the payload for
          this event from file */
        if (isFragmented(nbEvent)) {
            try {
                fragmentationHelper.refillPayloadIntoEvent(nbEvent);
            } catch (ServiceException serEx) {
                System.out.println(moduleName + serEx);
                System.out.println(moduleName +
                                   "No retransmission possible!!|n\n\n");
                return;
            }
        }

        RdCompanionEntityEvent companionEvent =
                (RdCompanionEntityEvent) pendingCompanions.get(eventId);

        RdRepublishedEntityEvent publishAgain =
                new RdRepublishedEntityEvent(nbEvent, companionEvent);
        templateId = nbEvent.getEventHeaders().getTemplateId();

        try {
            NBEvent nbEvent2 =
                    producer.generateEvent(TemplateProfileAndSynopsisTypes.
                                           STRING,
                                           rdsString + templateId,
                                           publishAgain.getBytes());

            NBEventGenerator.setEventType(nbEvent2,
                                          NBEventTypes.
                                          RELIABLE_DELIVERY_EXCHANGE);
            producer.publishEvent(nbEvent2);
        } catch (ServiceException serEx) {
            System.out.println(moduleName + serEx);
        }

    }


    private void
            processAcknowledgement(RdNegotiationsServiceEvent negotiationEvent) {
        int catenationAck = negotiationEvent.getAckCatenationNumber();
        int templateId = negotiationEvent.getTemplateId();
        if (negotiationEvent.isStandaloneAck()) {
            if (PublisherPercolator_Debug) {
                System.out.println("Received standalone acknowledgement for " +
                                   catenationAck);
            }
            processStandaloneAcknowledgement(templateId, catenationAck);
            return;
        }

        Object catenationKey = new Integer(catenationAck);
        Object templateKey = new Integer(negotiationEvent.getTemplateId());
        /** Check if ack was already processed */

        /* First check to see if an entry exists in acknowledgedCatenations for
           this templateKey, if not create one */
        if (!acknowledgedCatenations.containsKey(templateKey)) {
            acknowledgedCatenations.put(templateKey, new Integer(catenationAck));
            processAcknowledgementsUntil(templateId, catenationAck);
            return;
        }

        int lastAcknowledged =
                ((Integer) acknowledgedCatenations.get(templateKey)).intValue();

        if (lastAcknowledged >= catenationAck) {
            if (PublisherPercolator_Debug) {
                System.out.println(moduleName + "Catenation [" + catenationAck +
                                   "] was previously processed");
            }
            return;
        }

        /** Allow for the fact that acks may have been lost in transit. For a
          given templateId if ack5 then ack1-4 if they pertain to the same
          templateId */
        processAcknowledgementsUntil(templateId, catenationAck);
    }


    /** For a given templateId, process the acknowledgements up until
        acknowledgement number ackTill */
    private void processAcknowledgementsUntil(int templateId, int ackTill) {
        Object templateKey = new Integer(templateId);

        LinkedList _catenationChain = (LinkedList) templateCatenationChain.get(
                templateKey);

        int numOfCatenations = 0;
        for (int i = 0; i < _catenationChain.size(); i++) {
            int catValue = ((Integer) _catenationChain.get(i)).intValue();
            if (catValue > ackTill) {
                break;
            }
            numOfCatenations++;
        }
        /*end for */

        for (int i = 0; i < numOfCatenations; i++) {
            int _catenValue = ((Integer) _catenationChain.get(0)).
                              intValue();
            removePendingEntriesFor(_catenValue, templateId);
            _catenationChain.remove(0);
        }

        acknowledgedCatenations.remove(templateKey);
        acknowledgedCatenations.put(templateKey, new Integer(ackTill));
    }

    private void processStandaloneAcknowledgement(int templateId,
                                                  int standaloneAck) {
        Object templateKey = new Integer(templateId);

        LinkedList _catenationChain = (LinkedList) templateCatenationChain.get(
                templateKey);
        removePendingEntriesFor(standaloneAck, templateId);
        _catenationChain.remove(""+standaloneAck);

    }

    /** Removes buffered entries that pertain to a given catenation number */
    private void removePendingEntriesFor(int catenationNum, int templateId) {
        if (PublisherPercolator_Debug) {
            System.out.println(moduleName + "Removing entries for catenation [" +
                               catenationNum + "]");
        }
        Object catenationKey = "" + templateId + catenationNum;
        if (!catenationAndEventIds.containsKey(catenationKey)) {
            System.out.println(moduleName + "Catenation num [" + catenationNum +
                               "] missing!!. Was this processed before??. Rare!!");
            return;
        }

        Object eventId = catenationAndEventIds.get(catenationKey);

        pendingCompanions.remove(eventId);
        NBEvent nbEvent = (NBEvent) pendingEvents.remove(eventId);

        if (nbEvent != null) {
            if (isFragmented(nbEvent)) {
                try {
                    fragmentationHelper.fragmentStoredReliably(nbEvent);
                } catch (ServiceException serEx) {
                    System.out.println(moduleName + serEx);
                }
            }
        }

        catenationAndEventIds.remove(catenationKey);
    }


    public void processRecoveryResponse(RdRecoveryResponse recoveryResponse) {
        if (PublisherPercolator_Debug) {
            System.out.println(moduleName +
                               "Received recovery response with Publisher Info" +
                               recoveryResponse.getPublisherRecoveryInfo());
        }
        RdPublisherRecoveryInfo recoveryInfo =
                recoveryResponse.getPublisherRecoveryInfo();

        int templateId = recoveryInfo.getTemplateId();
        Object templateKey = new Integer(templateId);
        int catenation = recoveryInfo.getCatenationNumber();

        templatesAndCatenations.put(templateKey, new Integer(catenation));
        templateCatenationChain.put(templateKey, new LinkedList());
        int assignedCatenation = catenation;

        assignedCatenations.remove(templateKey);
        assignedCatenations.put(templateKey, new Integer(assignedCatenation));

        if (PublisherPercolator_Debug) {
            System.out.println(moduleName +
                               "CATENATION Number has been set to [" +
                               assignedCatenation + "]\n\n");
        }
        try {
            fragmentationHelper.onRecovery(assignedCatenation);
        } catch (ServiceException serEx) {
            System.out.println(moduleName + serEx);
        }
    }


    public void issueRecoveryRequest(RdRecoveryEntityRequest recoveryRequest) throws
            ServiceException {
        String subscription = entityId + "/Recovery/Publisher";
        int templateId = recoveryRequest.getTemplateId();

        NBEvent nbEvent =
                producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
                                       rdsString + templateId + "/Recovery",
                                       recoveryRequest.getBytes());

        NBEventGenerator.setEventType(nbEvent,
                                      NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);
        producer.publishEvent(nbEvent);
        System.out.println("Issued Recovery Request "+recoveryRequest);
    }


    private boolean isFragmented(NBEvent nbEvent) {
        if (!nbEvent.hasEventHeaders()) {
            return false;
        }

        EventHeaders eventHeaders = nbEvent.getEventHeaders();
        return eventHeaders.isFragmented();
    }

}
